package org.example.flatmap;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * map 算子
 */
public class DataFlatMapStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = env.fromElements("a,g,v", "a,h,j", "a,h,j", "a,h,j");
        source.flatMap(new MyMap()).print();
        env.execute();
    }

    static class MyMap implements FlatMapFunction<String, Object> {
        public void flatMap(String value, Collector<Object> out) throws Exception {
            String[] split = value.split(",");
            out.collect(split[2] + "2");
            out.collect(split[1] + "1");
        }
    }
}